package com.lizhiyu.flink.demo3_sink.mysql;

import com.lizhiyu.flink.model.Constant;
import com.lizhiyu.flink.model.VideoOrder;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Date;

/***
 * 使用官网自带链接插件向mysql中输入数据
 */
public class MysqlConnectorByFlink {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                new VideoOrder("1", "Stream ", 1, 2019,new Date()),
                new VideoOrder("1", "Stream ", 1, 2019,new Date()),
                new VideoOrder("1", "Streamk", 1, 2019,new Date()),
                new VideoOrder("1", "Stream", 1, 2019,new Date())
        ).addSink(
                JdbcSink.sink(
                        "INSERT INTO `video_order` (`user_id`, `money`, `title`, `trade_no`, `create_time`) VALUES(?,?,?,?,?)",
                        (statement, videoOrder) -> {
                            statement.setLong(1, videoOrder.getUserId());
                            statement.setLong(2, videoOrder.getMoney());
                            statement.setString(3, videoOrder.getTitle());
                            statement.setString(4, videoOrder.getTradeNo());
                            statement.setDate(5,  new java.sql.Date(videoOrder.getCreateTime().getTime()));
                        },
                        JdbcExecutionOptions.builder()
                                .withBatchSize(1000)
                                .withBatchIntervalMs(200)
                                .withMaxRetries(5)
                                .build(),

                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl(Constant.url)
                                .withDriverName(Constant.driver)
                                .withUsername(Constant.userName)
                                .withPassword(Constant.password)
                                .build()
                ));

        env.execute();



    }
}
